package com.example.demo.kafka;

import com.alibaba.fastjson.JSON;
import com.example.demo.config.CdcConfig;
import com.example.demo.dto.CdcMessage;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

@Service
public class CdcMessageConsumer {

    private static final Logger logger = LoggerFactory.getLogger(CdcMessageConsumer.class);
    @Autowired
    RestTemplate restTemplate;

    @Autowired
    private CdcConfig cdcConfig;

//    @KafkaListener(topics = "#{@cdcConfig.kafkaTopic}")
//    public void listen(ConsumerRecord<String, byte[]> record) {
//        doConsumer(record);
//    }

    public void doConsumer(ConsumerRecord<String, byte[]> record) {
        byte[] rawBytes = record.value();
        try {
            // 假设消息实际是 UTF-8 文本（如 JSON）
            String jsonString = new String(rawBytes, StandardCharsets.UTF_8);
            logger.info("收到消息: " + jsonString);
            // 调用Python脚本并传递JSON数据
//            invokePythonScript(jsonString);
            invokePython(jsonString);
        } catch (Exception e) {
            logger.error("无法解析，打印 hex: " + bytesToHex(rawBytes),e);
        }
    }

    public void invokePython(String content){
            // 构造 payload
            Map<String, Object> payload = new HashMap<>();
            payload.put("moduleId", "1945670208280907778");

            Map<String, Object> parameter = new HashMap<>();
            parameter.put("data", content);
            payload.put("parameter", parameter);
            // ... 根据实际内容添加更多键值

            // 设置 headers
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            headers.set("uid", "1");

            // 构造请求实体
            HttpEntity<Map<String, Object>> request = new HttpEntity<>(payload, headers);

            // 初始化 RestTemplate
            RestTemplate restTemplate = new RestTemplate();

            // 发送 POST 请求
            String url = "http://10.150.8.18:8910/py-land";
            ResponseEntity<String> response = restTemplate.postForEntity(url, request, String.class);

            // 输出响应
            System.out.println("响应状态码: " + response.getStatusCode());
            System.out.println("响应内容: " + response.getBody());
    }

    public static String bytesToHex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bytes) {
            sb.append(String.format("%02x", b));
        }
        return sb.toString();
    }
    private void invokePythonScript(String jsonString) throws IOException, InterruptedException {
        // 将消息转换为JSON字符串
//        String jsonString = convertToJsonString(message);
        logger.info("json数据:{}",jsonString);
        // 构建Python命令
        ProcessBuilder processBuilder = new ProcessBuilder(
            cdcConfig.getPythonCmdPath(), // Python解释器路径，根据环境调整
            cdcConfig.getPythonFilePath()
        );
        processBuilder.environment().put("PYTHONIOENCODING", "utf-8"); // 设置Python的I/O编码
        Process process = processBuilder.start();
        try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(process.getOutputStream()))) {
            writer.write(jsonString); // 直接写入原始JSON字符串
            writer.flush();
        }
        // 设置工作目录和环境变量（如有需要）
        // processBuilder.directory(new File("/path/to/working/directory"));

        // 获取Python脚本的输出
        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) {
            String line;
            while ((line = reader.readLine()) != null) {
                logger.info("Python脚本输出: {}", line);
            }
        }
        
        // 获取错误输出
        try (BufferedReader errorReader = new BufferedReader(
                new InputStreamReader(process.getErrorStream(), StandardCharsets.UTF_8))) {
            String line;
            StringBuilder errorMessage = new StringBuilder();
            while ((line = errorReader.readLine()) != null) {
                errorMessage.append(line).append("\n");
            }
            if (errorMessage.length() > 0) {
                logger.error("Python脚本错误: {}", errorMessage);
            }
        }
        
        // 等待进程完成并获取退出码
        int exitCode = process.waitFor();
        if (exitCode != 0) {
            logger.error("Python脚本执行失败，退出码: {}", exitCode);
        }
    }

    private String convertToJsonString(HashMap message) {
        // 实际项目中应使用JSON序列化工具，如Jackson或Gson
        // 这里简化处理，实际使用时请替换为正式实现
        return JSON.toJSONString(message);
    }

    private void processCdcMessage(HashMap message) {
        // 原有业务逻辑保持不变
        processInsert(message);
    }

    private void processInsert(HashMap message) {
//        logger.info("处理INSERT操作 - 表: {}, 数据: {}",
//                message.getMetadata().getTable(),
//                message.getAfter());
        // 插入业务逻辑
    }

    private void processUpdate(CdcMessage message) {
        logger.info("处理UPDATE操作 - 表: {}, 前值: {}, 后值: {}",
                message.getMetadata().getTable(),
                message.getBefore(),
                message.getAfter());
        // 更新业务逻辑
    }

    private void processDelete(CdcMessage message) {
        logger.info("处理DELETE操作 - 表: {}, 数据: {}",
                message.getMetadata().getTable(),
                message.getBefore());
        // 删除业务逻辑
    }
}    